package com.sinohealth.sca.listener;

import com.sinohealth.sca.model.PointMessage;
import com.sinohealth.sca.service.PointUploadService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Description: 埋点消息监听器
 * @Package: com.sinohealth.sca.listener
 * @Author: chenweifeng
 * @Date: 2024/04/07 上午11:32
 **/
@Slf4j
@Component
public class PointMessageListener {

    @Autowired
    private PointUploadService pointUploadService;

    // 使用阻塞队列来积累消息
    private final BlockingQueue<PointMessage> messageQueue = new LinkedBlockingQueue<>();

    // 批量上传的阈值，可以通过配置文件进行配置，如果未配置则默认为100
    @Value("${point.upload.batch.size:100}")
    private int batchSize;

    // 默认每隔3秒执行一次
    private final static int BATCH_SCHEDULED_TIME = 3000;


    // 异步处理消息
    @Async(value = "handleMessageThreadPool")
    public void handleMessage(PointMessage message) {
        try {
            // 将消息放入队列
            messageQueue.put(message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error(e.getMessage(), e);
        }

        // 检查队列大小，达到阈值则进行批量处理
        if (messageQueue.size() >= batchSize) {
            flushQueue();
        }
    }

    // 定时任务，定期上传积压的消息
    @Async("processQueueThreadPool")
    @Scheduled(fixedDelay = BATCH_SCHEDULED_TIME)
    public void processQueuePeriodically() {
        if (!messageQueue.isEmpty()) {
            flushQueue();
        }
    }

    // 批量上传消息
    private void flushQueue() {
        List<PointMessage> messageList = new ArrayList<>();
        messageQueue.drainTo(messageList, batchSize); // 从队列中取出一定数量的消息

        if (!messageList.isEmpty()) {
            pointUploadService.upload(messageList); // 调用上传服务
        }
    }
}
